package com.zyx.sparkdemo.streaming

import java.util.concurrent.Future

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}

/**
 * @author Yaxi.Zhang
 * @since 2021/9/14 20:29
 *        desc: KafkaSink的构建
 *        reference:  https://blog.csdn.net/u013013024/article/details/77877570
 *                    https://blog.csdn.net/qq_32253371/article/details/79308786?utm_medium=distribute.pc_relevant.none-task-blog-2
 *                    %7Edefault%7ECTRLIST%7Edefault-1.no_search_link&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2
 *                    %7Edefault%7ECTRLIST%7Edefault-1.no_search_link
 */
object KafkaSink {

  import scala.collection.JavaConversions._

  // 注入方法
  def apply[K, V](config: Map[String, Object]): KafkaSink[K, V] = {

    val createProducerFunc = () => {
      val producer = new KafkaProducer[K, V](config)
      sys.addShutdownHook {
        // Ensure that, on executor JVM shutdown, the Kafka producer sends
        // any buffered messages to Kafka before shutting down.
        producer.close()
      }
      producer
    }

    new KafkaSink(createProducerFunc)
  }

  // 注入方法重载
  def apply[K, V](config: java.util.Properties): KafkaSink[K, V] = apply(config.toMap)

}

class KafkaSink[K, V](createProducer: () => KafkaProducer[K, V]) extends Serializable {
  /* This is the key idea that allows us to work around running into
     NotSerializableExceptions. */
  lazy val producer: KafkaProducer[K, V] = createProducer()

  def send(topic: String, key: K, value: V): Future[RecordMetadata] =
    producer.send(new ProducerRecord[K, V](topic, key, value))

  def send(topic: String, value: V): Future[RecordMetadata] =
    producer.send(new ProducerRecord[K, V](topic, value))
}

